package io.mft.datasource.mq.rabbitmq.consumer;

import java.io.IOException;
import java.io.UnsupportedEncodingException;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;

import com.rabbitmq.client.Channel;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import io.mft.commons.config.mq.MqConsumer;
import io.mft.commons.config.mq.MqHandlerMode;
import io.mft.commons.function.ExecuteReturn;
import io.mft.commons.function.response.kv.KVMap;
import io.mft.datasource.mq.container.MessageProcessor;
import io.mft.datasource.mq.container.impl.MessageFuncNoProcessor;
import io.mft.datasource.mq.container.impl.MessageImplProcessor;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class MessageHandler  implements ChannelAwareMessageListener {

    private MqConsumer consumer;
    private MessageProcessor messageProcessor;

	public MessageHandler(MqConsumer consumer) {
		this.consumer = consumer;
		if (consumer.getHandlerMode() == MqHandlerMode.funcNo) {
			messageProcessor = new MessageFuncNoProcessor();
		} else {
			messageProcessor = new MessageImplProcessor();
		}
	}
    @Override
    public void onMessage(Message message , Channel channel) {
		try {
            MessageProperties messageProperties = message.getMessageProperties();
            String queueName =  messageProperties.getConsumerQueue() ;
    		byte[] bytes = message.getBody();
			String msg = new String(bytes, "UTF-8");
			log.warn("监听到消息-开始处理>>>队列:{}->>>消息体:{}",queueName,msg);
			JSONObject jsonObject=JSONUtil.parseObj(msg);
            
            KVMap kvMessage=new KVMap();
            kvMessage.putAll(jsonObject);
            ExecuteReturn executeReturn=messageProcessor.process(consumer, kvMessage);
            if (executeReturn.getCode()==0) {
            	 channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            	 log.warn("监听到消息-完成处理-ACK>>>队列:{}->>>消息体:{}",queueName,msg);
			}else {
				channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
				 log.warn("监听到消息-完成处理-NACK>>>队列:{}->>>消息体:{}",queueName,msg);
			}
           
		} catch (UnsupportedEncodingException e1) {
			try {
				channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
			} catch (IOException e) {
				log.error("", e);
			}
			log.error("", e1);
		} catch (IOException e) {
			try {
				channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
			} catch (IOException e1) {
				log.error("", e1);
			}
		};
    }
}
